package com.gome.han.bigdata.hbase.apacheHBaseAPIs._002demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Logger;

/**
 * @author Hanpeng
 * @date 2020/12/12 10:26
 * @description: HBase执行DMl操作
 */
public class TestDML {
    private static Connection conn  =null;
    private static Admin      admin =null;
    static {
        try {
            //1,获取配置信息
            Configuration config = HBaseConfiguration.create();
            //conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem")
            //config.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");//无论设置与否 都不会报错
            config.set("hbase.zookeeper.quorum", "elkfour");
            //2,创建连接对象
            conn = ConnectionFactory.createConnection(config);
            //创建admin 对象
            admin = conn.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    private static Logger logger = Logger.getLogger(TestDDL.class.getName());
    public static void main(String[] args) throws IOException {
        //putData("stu", "1006", "info", "name", "han2");
        //putDataSWithColunm("stu", "1006", "info", "name", "han2");
        //getData("stu", "1006", "", "");
        deleteData("stu","1006", "sas", "sd");
        close();
    }
    public  static  void putData(String tableName,String rowKey,String cf,String column,String value) {
        Table table=null;
        try {
            //1,创建表
            table = conn.getTable(TableName.valueOf(tableName));
            //2,创建put对象
            Put put = new Put(Bytes.toBytes(rowKey));
            //3,给put赋值
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(value));
            //4,插入数据
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if(table!=null) {
                try {
                    table.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * @param tableName
     * @param rowKey
     * @param cf
     * @param column
     * @param value
     * 插入多个列
     */
    public  static  void putDataSWithColunm(String tableName,String rowKey,String cf,String column,String value) {
        Table table=null;
        try {
            //1,创建表
            table = conn.getTable(TableName.valueOf(tableName));
            //2,创建put对象
            Put put = new Put(Bytes.toBytes(rowKey));
            //3,给put赋值
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(value));
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("sex"), Bytes.toBytes("male"));
            //4,插入数据
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if(table!=null) {
                try {
                    table.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * @param tableName
     * @param rowKey
     * @param cf
     * @param column
     * @param value
     * 插入数据有多个key
     */
    public  static  void putDatas(String tableName,String rowKey,String cf,String column,String value) {
        Table table=null;
        try {
            //1,创建表
            table = conn.getTable(TableName.valueOf(tableName));
            //2,创建put对象
            Put put = new Put(Bytes.toBytes(rowKey));
            //3,给put赋值
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(value));
            put.addColumn(Bytes.toBytes(cf), Bytes.toBytes("sex"), Bytes.toBytes("male"));
            //重复创建多个put对象 放入List集合之中
            ArrayList<Put> putsList = new ArrayList<Put>(600);
            //4,插入数据
            table.put(putsList);
            //生产环境中可以达到一定数目之后进行提交 并且用另外一条线程监控这条线程 如果经过一定时间 还没有提交 敦促它进行提交
            //
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if(table!=null) {
                try {
                    table.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 获取数据
     */
    public static void getData(String tableName, String rowKey, String cf, String column) {
        Table table = null;
        try {
            //1,创建表
            table = conn.getTable(TableName.valueOf(tableName));
            //2,创建get对象
            Get get = new Get(Bytes.toBytes(rowKey));
            //2.1 指定列族
            //get.addFamily(Bytes.toBytes(cf));
            //2.2 指定列
            //get.addColumn(Bytes.toBytes(cf),Bytes.toBytes(column));

            //2.3 指定版本
            //get.readAllVersions();
            //3,获取数据
            Result result = table.get(get);
            //解析result 并打印
            Cell[] cells = result.rawCells();
            if(cells!=null&&cells.length>0) {
                for (Cell cell:cells) {
                    /*System.out.println(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength())+"::"+
                            Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));*/
                    byte[] bytesCF = CellUtil.cloneFamily(cell);
                    byte[] bytesQualifier = CellUtil.cloneQualifier(cell);
                    byte[] bytesValue = CellUtil.cloneValue(cell);
                    byte[] bytesRow = CellUtil.cloneRow(cell);
                    long timestamp = cell.getTimestamp();
                    Cell.Type type = cell.getType();
                    System.out.println(Bytes.toString(bytesCF));
                    System.out.println(Bytes.toString(bytesQualifier));
                    System.out.println(Bytes.toString(bytesValue));
                }
            }

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(table!=null) {
                try {
                    table.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
    /**
     * 获取数据
     */
    public static void scanTable(String tableName) {
        Table table = null;
        try {
            //1,创建表
            table = conn.getTable(TableName.valueOf(tableName));
            //2,获取scan
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes("1003"));//设置开始扫描的rowK ey
            
            //貌似一次拉取一定数目的
            ResultScanner scanner = table.getScanner(scan);
            
            Iterator<Result> iterator = scanner.iterator();
            //
          /*  while (iterator.hasNext()) {
                Result next = iterator.next();
                //这里面我们就熟悉了
            }*/
            for (Result resultTemp:scanner) {
                for(Cell cellTemp:resultTemp.rawCells()) {
                    //这里我们也熟悉了
                   // cellTemp.getType();
                }
            }


        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(table!=null) {
                try {
                    table.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 可以根据 rowkey  cf 以及column进行删除
     * 如果保存多个版本 能否删除最新的或者最旧的版本捏?
     */
    public  static  void  deleteByRowKey() {

    }

    /**
     * @param tableName
     * @param rowKey
     * @param cf
     * @param column
     *
     * 根据rowkey 删除 标记是  1006  column=info:, timestamp=1607755883345, type=DeleteFamily 这一行将查询不到了
     *
     */
    public static  void  deleteData(String tableName,String rowKey,String cf,String column) {
        Table table = null;
        try {
            //1,创建表
            table = conn.getTable(TableName.valueOf(tableName));
            //2,构建删除命令对象
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            //2.1 设置删除的列 这个是删除最新版本的 可能会出现低版本的 但是flush 之后才不出现低版本的 这个生产环境中慎用
            /*put "t1","1008","info:name","123"
            put "t1","1008","info:name","234"
            put "t1","1008","info:name","345"
            flush
            然后执行 delete.addColumn 这时候就查询不到次新值
            */
            //rowkey   ===>DeleteFamily
            //rowkey+cf===>DeleteFamily
            //rowkey+cf+column+s(addColumns)==>DeleteColumn
            //rowkey+cf+column+不加s(addColumn)===>delete 删除特定条
            delete.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column));
           // //删除所有版本
            delete.addColumns(Bytes.toBytes(cf), Bytes.toBytes(column));
            //2.2 删除指定版本的
            delete.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), System.currentTimeMillis());
            //2.3 删除指定列族
            delete.addFamily(Bytes.toBytes(cf));
            table.delete(delete);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(table!=null) {
                try {
                    table.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
    private  static  void  close() {
        if(admin!=null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if(conn!=null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
